// ThreadPool.cc - 线程池实现
// 基于Muduo设计哲学，提供高效、灵活的线程池功能

#include "ThreadPool.h"
#include <atomic>
#include <chrono>
#include <iostream>
#include <stdexcept>
#include <string_view>

namespace cytoolkit {

// ThreadPool构造函数实现
ThreadPool::ThreadPool(const ThreadPoolOptions& options) 
    : impl_(std::make_unique<ThreadPoolImpl>(options)) {
    // 默认构造函数，线程池初始状态为kRunning
}

ThreadPool::ThreadPool(size_t numThreads, const std::string& name) {
    ThreadPoolOptions options;
    options.name = name;
    options.initialThreads = numThreads;
    impl_ = std::make_unique<ThreadPoolImpl>(options);
}

ThreadPool::~ThreadPool() {
    // 确保在析构时关闭线程池
    if (impl_->state() == ThreadPoolState::kRunning) {
        impl_->shutdown();
    }
}

// ThreadPool方法实现
void ThreadPool::start() {
    impl_->start();
}

void ThreadPool::shutdown() {
    impl_->shutdown(true); // 等待任务完成
}

void ThreadPool::shutdownNow() {
    impl_->shutdown(false); // 不等待任务完成
}

void ThreadPool::waitForTasks() {
    impl_->waitForTasks();
}

bool ThreadPool::resize(size_t numThreads) {
    return impl_->resize(numThreads);
}

size_t ThreadPool::threadCount() const {
    return impl_->threadCount();
}

size_t ThreadPool::activeThreadCount() const {
    return impl_->activeThreadCount();
}

size_t ThreadPool::queueSize() const {
    return impl_->queueSize();
}

ThreadPoolState ThreadPool::state() const {
    return impl_->state();
}

const std::string& ThreadPool::name() const {
    return impl_->name();
}

void ThreadPool::setThreadNamePrefix(const std::string& prefix) {
    impl_->setThreadNamePrefix(prefix);
}

void ThreadPool::clearTasks() {
    impl_->clearTasks();
}

bool ThreadPool::isFull() const {
    return impl_->isFull();
}

bool ThreadPool::isEmpty() const {
    return impl_->isEmpty();
}

// ThreadPoolImpl构造函数实现
ThreadPoolImpl::ThreadPoolImpl(const ThreadPoolOptions& options)
    : options_(options)
    , name_(options.name)
    , state_(ThreadPoolState::kRunning)
    , activeThreads_(0) {
    // 初始化线程池
}

ThreadPoolImpl::~ThreadPoolImpl() {
    // 确保在析构时关闭线程池
    if (state_ == ThreadPoolState::kRunning) {
        shutdown(true);
    }
}

// ThreadPoolImpl方法实现
void ThreadPoolImpl::start() {
    std::lock_guard<std::mutex> lock(mutex_);
    
    if (state_ != ThreadPoolState::kRunning) {
        throw std::runtime_error("ThreadPool already stopped");
    }
    
    // 创建工作线程
    workers_.reserve(options_.initialThreads);
    for (size_t i = 0; i < options_.initialThreads; ++i) {
        std::string threadName = name_;
        if (!threadNamePrefix_.empty()) {
            threadName = threadNamePrefix_ + "-" + std::to_string(i);
        } else {
            threadName = name_ + "-" + std::to_string(i);
        }
        
        workers_.emplace_back(new Thread(
            std::bind(&ThreadPoolImpl::workerThreadFunc, this),
            threadName
        ));
        
        workers_.back()->start();
    }
}

void ThreadPoolImpl::shutdown(bool waitForTasks) {
    {
        std::lock_guard<std::mutex> lock(mutex_);
        
        if (state_ != ThreadPoolState::kRunning) {
            return; // 已经停止
        }
        
        // 设置线程池状态为关闭中
        state_ = ThreadPoolState::kShuttingDown;
        
        // 如果不需要等待任务完成，则清空任务队列
        if (!waitForTasks) {
            clearTasks();
        }
    }
    
    // 通知所有等待的线程
    notEmpty_.notify_all();
    
    // 如果需要等待任务完成，则等待所有工作线程结束
    if (waitForTasks) {
        for (auto& worker : workers_) {
            if (worker && worker->joinable()) {
                worker->join();
            }
        }
    }
    
    // 设置状态为已终止
    state_ = ThreadPoolState::kTerminated;
    
    // 通知等待任务完成的线程
    taskDone_.notify_all();
}

void ThreadPoolImpl::waitForTasks() {
    std::unique_lock<std::mutex> lock(mutex_);
    
    // 等待任务队列为空且没有活跃线程
    taskDone_.wait(lock, [this]() {
        return tasks_.empty() && activeThreads_ == 0;
    });
}

bool ThreadPoolImpl::resize(size_t numThreads) {
    std::lock_guard<std::mutex> lock(mutex_);
    
    // 检查线程池状态
    if (state_ != ThreadPoolState::kRunning) {
        return false;
    }
    
    // 检查是否超过最大线程数限制
    if (options_.maxThreads > 0 && numThreads > options_.maxThreads) {
        return false;
    }
    
    size_t currentSize = workers_.size();
    
    // 增加线程
    if (numThreads > currentSize) {
        workers_.reserve(numThreads);
        // 修复循环条件，确保创建正确数量的线程
        for (size_t i = currentSize; i < numThreads; ++i) {
            std::string threadName = name_;
            if (!threadNamePrefix_.empty()) {
                threadName = threadNamePrefix_ + "-" + std::to_string(i);
            } else {
                threadName = name_ + "-" + std::to_string(i);
            }
            
            workers_.emplace_back(new Thread(
                std::bind(&ThreadPoolImpl::workerThreadFunc, this),
                threadName
            ));
            
            workers_.back()->start();
        }
    }
    // 注意：不支持减少线程数量，因为这会带来额外的复杂性
    // 可以通过allowCoreThreadTimeout和keepAliveTimeMs来实现
    
    return true;
}

void ThreadPoolImpl::workerThreadFunc() {
    // 线程名称已经在创建线程时通过Thread类设置
    // 这里可以添加其他线程初始化逻辑
    
    bool timeout = false;
    
    while (true) {
        std::function<void()> task;
        
        {  
            std::unique_lock<std::mutex> lock(mutex_);
            
            // 如果允许核心线程超时退出且队列中有空闲时间
            if (options_.allowCoreThreadTimeout) {
                // 等待任务或超时
                auto waitResult = notEmpty_.wait_for(lock, 
                    std::chrono::milliseconds(options_.keepAliveTimeMs),
                    [this]() {
                        return !tasks_.empty() || 
                               state_ != ThreadPoolState::kRunning;
                    });
                
                // 如果超时且队列为空，则退出
                if (!waitResult && tasks_.empty()) {
                    timeout = true;
                    break;
                }
            } else {
                // 一直等待直到有任务或线程池关闭
                notEmpty_.wait(lock, [this]() {
                    return !tasks_.empty() || 
                           state_ != ThreadPoolState::kRunning;
                });
            }
            
            // 线程池已关闭，退出线程
            if (state_ == ThreadPoolState::kTerminated || 
                (state_ == ThreadPoolState::kShuttingDown && tasks_.empty())) {
                break;
            }
            
            // 从队列中取出一个任务
            task = std::move(tasks_.front());
            tasks_.pop();
            
            // 通知等待的生产者
            notFull_.notify_one();
        }
        
        // 执行任务前增加活跃线程计数
        ++activeThreads_;
        
        try {
            // 执行任务，但在执行前检查线程池状态
            // 对于shutdownNow，我们不应该继续执行新任务
            if (state_ != ThreadPoolState::kRunning) {
                --activeThreads_;
                taskDone_.notify_all();
                break;
            }
            
            // 执行任务
            task();
        } catch (const std::exception& e) {
            // 记录异常
            std::cerr << "Exception in ThreadPool task: " << e.what() << std::endl;
        } catch (...) {
            // 记录未知异常
            std::cerr << "Unknown exception in ThreadPool task" << std::endl;
        }
        
        // 执行任务后减少活跃线程计数
        --activeThreads_;
        
        // 通知等待任务完成的线程
        taskDone_.notify_all();
    }
    
    // 如果是超时退出，从工作线程列表中移除
    if (timeout && state_ == ThreadPoolState::kRunning) {
        std::lock_guard<std::mutex> lock(mutex_);
        // 注意：由于使用了std::unique_ptr，这里需要更复杂的逻辑来移除已退出的线程
        // 简化处理，不实际从向量中移除，让析构函数处理
    }
}

size_t ThreadPoolImpl::threadCount() const {
    std::lock_guard<std::mutex> lock(mutex_);
    // 只计算有效的线程
    size_t count = 0;
    for (const auto& worker : workers_) {
        if (worker) {
            ++count;
        }
    }
    return count;
}

size_t ThreadPoolImpl::activeThreadCount() const {
    return activeThreads_;
}

size_t ThreadPoolImpl::queueSize() const {
    std::lock_guard<std::mutex> lock(mutex_);
    return tasks_.size();
}

ThreadPoolState ThreadPoolImpl::state() const {
    return state_;
}

const std::string& ThreadPoolImpl::name() const {
    return name_;
}

void ThreadPoolImpl::setThreadNamePrefix(const std::string& prefix) {
    threadNamePrefix_ = prefix;
}

void ThreadPoolImpl::clearTasks() {
    std::lock_guard<std::mutex> lock(mutex_);
    std::queue<std::function<void()>> empty;
    std::swap(tasks_, empty);
}

bool ThreadPoolImpl::isFull() const {
    std::lock_guard<std::mutex> lock(mutex_);
    return options_.maxQueueSize > 0 && tasks_.size() >= options_.maxQueueSize;
}

bool ThreadPoolImpl::isEmpty() const {
    std::lock_guard<std::mutex> lock(mutex_);
    return tasks_.empty();
}

} // namespace cytoolkit